package demand

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import util.JDBCUtil

import java.text.SimpleDateFormat
import java.util.Date
import scala.collection.mutable.ListBuffer

object Req011 {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[*]").setAppName("black_list")
    val ssc = new StreamingContext(conf, Seconds(3))

    val kafkaPara: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop-01:9092, hadoop-02:9092, hadoop-03:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "sparkStreaming",
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
    )
    val kafkaDataDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](Set("streamingData"), kafkaPara)
    )

    val userClickData = kafkaDataDS.map(
      kafkaData => {
        val data = kafkaData.value()
        val datas = data.split(" ")
        AdClickData(datas(0), datas(1), datas(2), datas(3), datas(4))
      }
    )

    val ds: DStream[((String, String, String), Int)] = userClickData.transform(
      rdd => {
        val blackList = ListBuffer[String]()

        val connection = JDBCUtil.getConnection
        val pstat = connection.prepareStatement("select userid from black_list")

        val result = pstat.executeQuery()
        while (result.next()) {
          blackList.append(result.getString(1))
        }

        result.close()
        pstat.close()
        connection.close()

        val filterRDD = rdd.filter(
          data => {
            !blackList.contains(data.user)
          }
        )
        filterRDD.map(
          data => {
            val sdf = new SimpleDateFormat("yyyy-MM-dd")
            val day = sdf.format(new Date(data.ts.toLong))
            val user = data.user
            val ad = data.ad
            ((day, user, ad), 1)
          }
        ).reduceByKey(_ + _)
      }
    )

    ds.foreachRDD(
      rdd => {

        rdd.foreachPartition(

          iter => {
            val conn = JDBCUtil.getConnection
            iter.foreach{

              case ((day, user, ad), count) => {
                println(s"${day} ${user} ${ad} ${count}")
                if (count >= 3) {
                  val sql =
                    """
                      |insert into black_list (userid) values (?)
                      |on DUPLICATE KEY
                      |UPDATE userid = ?
                      |""".stripMargin
                  JDBCUtil.executeUpdate(conn, sql, Array(user, user))
                }else{
                  val sql =
                    """
                      |select *
                      |from user_ad_count
                      |where dt = ? and userid = ? and adid = ?
                      |""".stripMargin
                  val flag = JDBCUtil.isExist(conn, sql, Array(day, user, ad))
                  if (flag) {
                    val sql1 =
                      """
                        |updata user_ad_count
                        |set count = count + ?
                        |where dt = ? and userid = ? and adid = ?
                        |""".stripMargin
                    JDBCUtil.executeUpdate(conn, sql1, Array(day, user, ad))
                    //更新数据后再次进行判断
                    val sql2 =
                      """
                        |select *
                        |from user_ad_count
                        |where dt = ? and userid = ? and adid = ? and count ?= 30
                        |""".stripMargin
                    val flag1 = JDBCUtil.isExist(conn, sql2, Array(day, user, ad))
                    if (flag1) {
                      val sql3 =
                      """
                         |insert into black_list (userid) values (?)
                         |on DUPLICATE KEY
                         |UPDATE userid = ?
                         |""".stripMargin
                      JDBCUtil.executeUpdate(conn, sql3, Array( user, user ))
                    }
                  }else {
                    val sql4 = """
                                 | insert into user_ad_count ( dt, userid, adid, count ) values ( ?, ?, ?, ? )
                                           """.stripMargin
                    JDBCUtil.executeUpdate(conn, sql4, Array( day, user, ad, count ))
                  }
                }
              }
            }
            conn.close()
          }
        )
      }
    )

    ssc.start()
    ssc.awaitTermination()
  }
  case class AdClickData(ts: String, area: String, city: String, user: String, ad: String)
}
